import kafka.serializer.StringDecoder
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf

object DirectKafkaWordCount
{
def main(args: Array[String])
{

  val Array(brokers, topics) = args

  val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
  val ssc = new StreamingContext(sparkConf, Seconds(5))
  //kafka的topic集合，即可以订阅多个topic，args传参的时候用,隔开
  val topicsSet = topics.split(",").toSet
  //设置kafka参数，定义brokers集合
  val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
  val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]
  (
    ssc, kafkaParams, topicsSet)
  print("---------:" +messages)

  val lines = messages.map(_._2)
  val words = lines.flatMap(_.split(" "))
  val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
  wordCounts.print()

  ssc.start()
  ssc.awaitTermination()
 }
}

//链接：https://www.jianshu.com/p/e285882d2845
